package com.flying.rsocket;

import io.rsocket.AbstractRSocket;
import io.rsocket.Payload;
import io.rsocket.RSocketFactory;
import io.rsocket.transport.netty.server.TcpServerTransport;
import lombok.extern.slf4j.Slf4j;
import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
import reactor.core.Disposable;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import static com.flying.rsocket.Constants.TCP_PORT;

@Slf4j
public class Server {
    private final Disposable server;
    private final DataPublisher dataPublisher = new DataPublisher();
    private final GameController gameController;

    public Server() {
        this.server = RSocketFactory.receive()
                .acceptor((setup, sendingSocket) -> Mono.just(new RSocketImpl()))
                .transport(TcpServerTransport.create("localhost", TCP_PORT))
                .start()
                .subscribe();
        this.gameController = new GameController("Server Player");
    }

    public void dispose() {
        dataPublisher.complete();
        this.server.dispose();
    }

    private class RSocketImpl extends AbstractRSocket {
        @Override
        public Mono<Payload> requestResponse(Payload payload) {
            try {
                return Mono.just(payload);
            } catch (Exception e) {
                return Mono.error(e);
            }
        }

        @Override
        public Mono<Void> fireAndForget(Payload payload) {
            try {
                dataPublisher.publish(payload);
                return Mono.empty();
            } catch (Exception e) {
                return Mono.error(e);
            }
        }

        @Override
        public Flux<Payload> requestStream(Payload payload) {
            return Flux.from(dataPublisher);
        }

        @Override
        public Flux<Payload> requestChannel(Publisher<Payload> payloads) {
            Flux.from(payloads).subscribe(gameController::processPayload);
            return Flux.from(gameController);
        }
    }

    private class DataPublisher implements Publisher<Payload> {
        private Subscriber<? super Payload> subscriber;

        @Override
        public void subscribe(Subscriber<? super Payload> s) {
            this.subscriber = s;
        }

        public void publish(Payload payload) {
            if (subscriber != null) {
                subscriber.onNext(payload);
            }
        }

        public void complete() {
            if (subscriber != null) {
                subscriber.onComplete();
            }
        }
    }
}
